package net.bwie.jtp.dws.log.Function;

import net.bwie.jtp.dws.log.bean.PageViewBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class PageViewWindowFunction implements WindowFunction<PageViewBean, String, String, TimeWindow> {


    private FastDateFormat  format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void apply(String s, TimeWindow window, Iterable<PageViewBean> input, Collector<String> out) throws Exception {

        //1. 获取窗口开始和结束时间
        String windowStartTime = format.format(window.getStart());
        String windowEndTime = format.format(window.getEnd());


        //2对窗口中的数据进行计算

        Long pvCount = 0L;
        Long pvDuringTime = 0L;
        Long uvCount = 0L;
        Long sessionCount = 0L;

        for (PageViewBean PageViewBean : input){
            pvCount += PageViewBean.getPvCount();
            pvDuringTime += PageViewBean.getPvDuringTime();
            uvCount += PageViewBean.getUvCount();
            sessionCount += PageViewBean.getSessionCount();
        }

        //3 输出
        String output = windowStartTime + "," + windowEndTime + ","
                + s + "," + pvCount + "," + pvDuringTime + "," + uvCount + "," + sessionCount + ","
                + System.currentTimeMillis();


        out.collect(output);












    }
}
